﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Threading;
using UtilZ.Dotnet.Ex.Base;
using UtilZ.Dotnet.Ex.Communication.Base;
using UtilZ.Dotnet.Ex.Communication.Net.Base;
using UtilZ.Dotnet.Ex.Communication.Net.Tcp;
using UtilZ.Dotnet.Ex.Communication.Net.Udp;
using UtilZ.Dotnet.Ex.Model;
using UtilZ.Dotnet.Ex.ZLog;

namespace UtilZ.Dotnet.Ex.Communication.Net.BMQ
{
    /// <summary>
    /// BMQ发布器
    /// </summary>
    public class BMQPublisher : IDisposable
    {
        private readonly BMQPublisherConfig _config;
        /// <summary>
        /// 获取配置
        /// </summary>
        public BMQPublisherConfig Config
        {
            get { return _config; }
        }

        private readonly ObjectEventWaitHandleManager _waitHandleManager;
        private readonly IUnpacker _unpacker;
        private readonly object _lock = new object();


        /// <summary>
        /// udp-client
        /// </summary>
        private readonly NetUdpClient _udpClient = null;

        /// <summary>
        /// key:主题;value:主题对应的订阅客户端EndPoint列表
        /// </summary>
        private readonly Dictionary<short, HashSet<EndPoint>> _udpSubTopicEndPointDic = null;

        /// <summary>
        /// key:客户端EndPoint;value:客户端已订阅的订阅列表
        /// </summary>
        private readonly Dictionary<EndPoint, HashSet<short>> _udpEndPointSubTopicDic = null;



        /// <summary>
        /// tcp连接监听器
        /// </summary>
        private readonly NetTcpListener _netTcpListener = null;

        /// <summary>
        /// key:主题;value:主题对应的订阅客户端NetTcpClient列表
        /// </summary>
        private readonly Dictionary<short, HashSet<IClient>> _tcpSubTopicClientDic = null;

        /// <summary>
        /// key:客户端NetTcpClient;value:客户端已订阅的订阅列表
        /// </summary>
        private readonly Dictionary<IClient, HashSet<short>> _tcpClientSubTopicDic = null;




        private readonly ThreadEx _heartThread = null;
        private readonly ThreadEx _pushThread = null;
        private readonly AutoResetEvent _messageEnqueueHandler = new AutoResetEvent(false);
        private readonly AutoResetEvent _messageDequeueHandler = new AutoResetEvent(false);
        private readonly ConcurrentQueue<IBMQDataMessage> _messageQueue = new ConcurrentQueue<IBMQDataMessage>();

        /// <summary>
        /// 推送完成回调
        /// </summary>
        public Action<IBMQDataMessage> PushCompleted;

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="config">配置</param>

        public BMQPublisher(BMQPublisherConfig config)
        {
            if (config == null)
            {
                throw new ArgumentNullException(nameof(config));
            }
            this._config = config;
            this._waitHandleManager = new ObjectEventWaitHandleManager();

            switch (config.Protocol)
            {
                case NetProtocols.Tcp:
                    this._unpacker = new BMQTcpTLVUnpacker();
                    this._tcpSubTopicClientDic = new Dictionary<short, HashSet<IClient>>();
                    this._tcpClientSubTopicDic = new Dictionary<IClient, HashSet<short>>();
                    var initPara = new NetTcpListenerInitPara(config.BindEndPoint, config.Backlog);
                    //initPara.AcceptMode = ThreadMode.Sync;
                    //initPara.SyncThreadName = $"BMQ{NetHelper.GetIPPortString(config.BindEndPoint)}监听线程";
                    this._netTcpListener = new NetTcpListener(this.AcceptSocketCallback, initPara);
                    this._netTcpListener.Start();
                    break;
                case NetProtocols.Udp:
                    this._unpacker = new BMQUDPUnpacker();
                    this._udpSubTopicEndPointDic = new Dictionary<short, HashSet<EndPoint>>();
                    this._udpEndPointSubTopicDic = new Dictionary<EndPoint, HashSet<short>>();
                    var udpIntiPara = new NetUdpClientInitPara(this._unpacker, config.BindEndPoint);
                    var udpClient = new NetUdpClient(udpIntiPara);
                    udpClient.Start();
                    this._udpClient = udpClient;
                    break;
                default:
                    throw new NotSupportedException($"不支持协议{config.Protocol.ToString()}");
            }

            this._unpacker.PackOutput += this.Process_PackOutput;

            this._heartThread = new ThreadEx(this.HeartThreadMethod, "BMQPublisher心跳线程");
            this._heartThread.Start();

            this._pushThread = new ThreadEx(this.PushThreadMethod, $"BMQ{NetHelper.GetIPPortString(config.BindEndPoint)}推送线程");
            this._pushThread.Start();
        }


        private void HeartThreadMethod(ThreadPara threadPara)
        {
            try
            {
                while (!threadPara.Token.IsCancellationRequested)
                {
                    try
                    {
                        if (threadPara.WaitOne(this._config.HeartIntervalMilliseconds))
                        {
                            continue;
                        }

                        switch (this._config.Protocol)
                        {
                            case NetProtocols.Tcp:
                                this.TcpHeart();
                                break;
                            case NetProtocols.Udp:
                                this.UdpHeart();
                                break;
                            default:
                                throw new NotSupportedException($"不支持协议{this._config.Protocol.ToString()}");
                        }
                    }
                    catch (Exception exi)
                    {
                        ZLoger.Error(exi, "BQM订阅器心跳线程异常");
                    }
                }
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex, "BQM订阅器心跳线程方法异常");
            }
        }

        private void UdpHeart()
        {
            EndPoint[] subEndPoints;
            lock (this._lock)
            {
                subEndPoints = this._udpEndPointSubTopicDic.Keys.ToArray();
            }

            BMQHeartResMessage heartResMessage;
            List<EndPoint> timeoutEndPointList = null;
            foreach (EndPoint subEndPoint in subEndPoints)
            {
                try
                {
                    var heartReqMessage = new BMQHeartReqMessage(TimeEx.GetTimestamp());
                    heartResMessage = (BMQHeartResMessage)this.SendSessionMesage(this._udpClient, heartReqMessage, subEndPoint);
                }
                catch (TimeoutException tex)
                {
                    //心跳超时，添加到取消订阅列表
                    ZLoger.Warn(tex);
                    if (timeoutEndPointList == null)
                    {
                        timeoutEndPointList = new List<EndPoint>();
                    }
                    timeoutEndPointList.Add(subEndPoint);
                }
                catch (Exception ex)
                {
                    ZLoger.Error(ex);
                }
            }

            if (timeoutEndPointList != null)
            {
                HashSet<short> subTopics;
                HashSet<EndPoint> topicSubEndPoints;
                lock (this._lock)
                {
                    foreach (EndPoint subEndPoint in timeoutEndPointList)
                    {
                        if (this._udpEndPointSubTopicDic.TryGetValue(subEndPoint, out subTopics))
                        {
                            this._udpEndPointSubTopicDic.Remove(subEndPoint);

                            foreach (var topic in subTopics)
                            {
                                if (this._udpSubTopicEndPointDic.TryGetValue(topic, out topicSubEndPoints))
                                {
                                    topicSubEndPoints.Remove(subEndPoint);
                                }
                            }
                        }
                    }
                }
            }
        }

        private void TcpHeart()
        {
            IClient[] clients;
            lock (this._lock)
            {
                clients = this._tcpClientSubTopicDic.Keys.ToArray();
            }


            List<IClient> timeoutClientList = null;
            BMQHeartResMessage heartResMessage;
            foreach (var client in clients)
            {
                try
                {
                    var heartReqMessage = new BMQHeartReqMessage(TimeEx.GetTimestamp());
                    heartResMessage = (BMQHeartResMessage)this.SendSessionMesage(client, heartReqMessage, ((NetTcpClient)client).LinkInfo.RemoteEndPoint);
                }
                catch (TimeoutException tex)
                {
                    //心跳超时，添加到取消订阅列表
                    ZLoger.Warn(tex);
                    if (timeoutClientList == null)
                    {
                        timeoutClientList = new List<IClient>();
                    }
                    timeoutClientList.Add(client);
                }
                catch (Exception ex)
                {
                    ZLoger.Error(ex);
                }
            }

            if (timeoutClientList != null)
            {
                HashSet<short> clientSubTopics;
                HashSet<IClient> topicSubClients;
                lock (this._lock)
                {
                    foreach (var client in timeoutClientList)
                    {
                        if (this._tcpClientSubTopicDic.TryGetValue(client, out clientSubTopics))
                        {
                            this._tcpClientSubTopicDic.Remove(client);

                            foreach (var topic in clientSubTopics)
                            {
                                if (this._tcpSubTopicClientDic.TryGetValue(topic, out topicSubClients))
                                {
                                    topicSubClients.Remove(client);
                                }
                            }
                        }
                    }
                }
            }
        }




        private void AcceptSocketCallback(AcceptSocketArgs args)
        {
            try
            {
                var tcpIntiPara = new NetTcpClientInitPara(this._unpacker, args.Client);
                var tcpClient = new NetTcpClient(tcpIntiPara);
                tcpClient.ConnectChanged += TcpClient_ConnectChanged;
                lock (this._lock)
                {
                    this._tcpClientSubTopicDic.Add(tcpClient, new HashSet<short>());
                }
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex);
            }
        }

        private void TcpClient_ConnectChanged(object sender, NetConnectedChangedArgs e)
        {
            if (e.Status == ConnectionStatus.Opened)
            {
                ZLoger.Info("BMQ-tcp连接已建立...");
            }
            else
            {
                ZLoger.Warn("BMQ-tcp连接已断开...");
                ((NetTcpClient)sender).ConnectChanged -= TcpClient_ConnectChanged;
                var client = (IClient)sender;
                HashSet<short> subTopicList;
                HashSet<IClient> tcpClientList;

                lock (this._lock)
                {
                    if (this._tcpClientSubTopicDic.TryGetValue(client, out subTopicList))
                    {
                        foreach (var topic in subTopicList)
                        {
                            if (this._tcpSubTopicClientDic.TryGetValue(topic, out tcpClientList))
                            {
                                tcpClientList.Remove(client);
                            }
                        }

                        this._tcpClientSubTopicDic.Remove(client);
                    }
                }
            }
        }

        private void Process_PackOutput(object sender, DataOutputArgs e)
        {
            try
            {
                var client = (IClient)sender;
                var args = (BMQDataOutputArgs)e;
                EndPoint senderEndPoint = ((NetLinkInfo)e.LinkInfo).RemoteEndPoint;
                //byte[] buffer = e.Data.Bytes;
                //byte messageType = buffer.First();
                byte messageType = args.Data2.First();
                switch (messageType)
                {
                    case BMQMessageTypes.SubReq:
                        ZLoger.Debug("收到订阅请求...");
                        var subReqMessage = new BMQSubReqMessage();
                        subReqMessage.FromBytes(args.Data2);
                        this.ProcessSubReqMesage(subReqMessage, client, senderEndPoint);
                        ZLoger.Debug("处理完成订阅请求...");
                        break;
                    case BMQMessageTypes.UnsubReq:
                        ZLoger.Debug("收到取消订阅请求...");
                        var unsubReqMessage = new BMQUnsubReqMessage();
                        unsubReqMessage.FromBytes(args.Data2);
                        this.ProcessUnsubReqMesage(unsubReqMessage, client, senderEndPoint);
                        ZLoger.Debug("处理完成取消订阅请求...");
                        break;
                    case BMQMessageTypes.HeartReq:
                        ZLoger.Debug("收到心跳请求...");
                        //收到心跳消息解析并回复
                        var heartReqMessage = new BMQHeartReqMessage();
                        heartReqMessage.FromBytes(args.Data2);
                        var heartResMessage = new BMQHeartResMessage(heartReqMessage.ContextId, heartReqMessage.Timestamp, TimeEx.GetTimestamp());
                        this.SendMesage(client, heartResMessage, senderEndPoint);
                        ZLoger.Debug("处理完成心跳请求...");
                        break;
                    case BMQMessageTypes.HeartRes:
                        ZLoger.Debug("收到心跳响应请求...");
                        var heartResMessage2 = new BMQHeartResMessage();
                        heartResMessage2.FromBytes(args.Data2);
                        EventWaitHandleInfo waitHandleInfo = this._waitHandleManager.GetEventWaitHandleInfo(heartResMessage2.ContextId);
                        if (waitHandleInfo != null)
                        {
                            waitHandleInfo.Set(heartResMessage2);
                        }
                        else
                        {
                            ZLoger.Warn($"收到超时的消息,消息类型:{nameof(BMQHeartResMessage)},传话标识:{heartResMessage2.ContextId}");
                        }
                        ZLoger.Debug("处理完成心跳响应请求...");
                        break;
                    default:
                        ZLoger.Warn($"未处理的BMQ消息类型\"{messageType}\"");
                        break;
                }
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex, "处理BMQ消息异常");
            }
        }

        private void ProcessUnsubReqMesage(BMQUnsubReqMessage unsubReqMessage, IClient client, EndPoint senderEndPoint)
        {
            lock (this._lock)
            {
                switch (this._config.Protocol)
                {
                    case NetProtocols.Tcp:
                        HashSet<IClient> tcpClientList;
                        if (this._tcpSubTopicClientDic.TryGetValue(unsubReqMessage.Topic, out tcpClientList))
                        {
                            tcpClientList.Remove(client);
                        }

                        this._tcpClientSubTopicDic.Remove(client);
                        break;
                    case NetProtocols.Udp:
                        HashSet<EndPoint> udpClientEndPointList;
                        if (this._udpSubTopicEndPointDic.TryGetValue(unsubReqMessage.Topic, out udpClientEndPointList))
                        {
                            udpClientEndPointList.Remove(senderEndPoint);
                        }

                        this._udpEndPointSubTopicDic.Remove((senderEndPoint));
                        break;
                    default:
                        throw new NotSupportedException($"不支持协议{this._config.Protocol.ToString()}");
                }

                var unsubResMessage = new BMQUnsubResMessage(unsubReqMessage.ContextId, 1);
                this.SendMesage(client, unsubResMessage, senderEndPoint);
            }
        }

        private void ProcessSubReqMesage(BMQSubReqMessage subReqMessage, IClient client, EndPoint senderEndPoint)
        {
            lock (this._lock)
            {
                HashSet<short> subTopicList;

                switch (this._config.Protocol)
                {
                    case NetProtocols.Tcp:
                        HashSet<IClient> tcpClientList;
                        if (!this._tcpSubTopicClientDic.TryGetValue(subReqMessage.Topic, out tcpClientList))
                        {
                            tcpClientList = new HashSet<IClient>();
                            this._tcpSubTopicClientDic.Add(subReqMessage.Topic, tcpClientList);
                        }
                        tcpClientList.Add(client);

                        if (!this._tcpClientSubTopicDic.TryGetValue(client, out subTopicList))
                        {
                            subTopicList = new HashSet<short>();
                            this._tcpClientSubTopicDic.Add(client, subTopicList);
                        }
                        subTopicList.Add(subReqMessage.Topic);
                        break;
                    case NetProtocols.Udp:
                        HashSet<EndPoint> udpClientEndPointList;
                        if (!this._udpSubTopicEndPointDic.TryGetValue(subReqMessage.Topic, out udpClientEndPointList))
                        {
                            udpClientEndPointList = new HashSet<EndPoint>();
                            this._udpSubTopicEndPointDic.Add(subReqMessage.Topic, udpClientEndPointList);
                        }
                        udpClientEndPointList.Add(senderEndPoint);

                        if (!this._udpEndPointSubTopicDic.TryGetValue(senderEndPoint, out subTopicList))
                        {
                            subTopicList = new HashSet<short>();
                            this._udpEndPointSubTopicDic.Add(senderEndPoint, subTopicList);
                        }
                        subTopicList.Add(subReqMessage.Topic);
                        break;
                    default:
                        throw new NotSupportedException($"不支持协议{this._config.Protocol.ToString()}");
                }

                var unsubResMessage = new BMQSubResMessage(subReqMessage.ContextId, 1);
                this.SendMesage(client, unsubResMessage, senderEndPoint);
            }
        }

        private readonly ArraySpan<byte> _sendMesageBuffer = new ArraySpan<byte>(new byte[8192]);
        private void SendMesage(IClient client, IBMQMessage message, EndPoint remoteEndPoint)
        {
            switch (this._config.Protocol)
            {
                case NetProtocols.Tcp:
                    message.ToTLVBytes(this._sendMesageBuffer);
                    ((NetTcpClient)client).Send(this._sendMesageBuffer.Array, this._sendMesageBuffer.Offset, this._sendMesageBuffer.Length);
                    break;
                case NetProtocols.Udp:
                    message.ToBytes(this._sendMesageBuffer);
                    ((NetUdpClient)client).SendTo(this._sendMesageBuffer.Array, this._sendMesageBuffer.Offset, this._sendMesageBuffer.Length, System.Net.Sockets.SocketFlags.None, remoteEndPoint);
                    break;
                default:
                    throw new NotSupportedException($"不支持协议{this._config.Protocol.ToString()}");
            }
        }


        private IBMQSessionMessage SendSessionMesage(IClient client, IBMQSessionMessage message, EndPoint remoteEndPoint)
        {
            int timeoutMilliseconds = this._config.TimeoutMilliseconds;
            //100是为确定超时后，一定不会处理收到的消息
            using (EventWaitHandleInfo waitHandleInfo = this._waitHandleManager.Create(message.ContextId, timeoutMilliseconds + 100))
            {
                this.SendMesage(client, message, remoteEndPoint);
                if (waitHandleInfo.WaitOne(timeoutMilliseconds))
                {
                    this._waitHandleManager.RemoveEventWaitHandle(message.ContextId);
                    return (IBMQSessionMessage)waitHandleInfo.Tag;
                }
                else
                {
                    throw new ApplicationException($"BMQ传话消息类型\"{message.GetType().FullName}\"传话\"{message.ContextId}\"响应超时...");
                }
            }
        }







        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="message">消息</param>
        /// <exception cref="ObjectDisposedException">对象释放将抛出异常</exception>
        public void Push(IBMQDataMessage message)
        {
            while (this._config.CacheQueueMax > 0 && this._messageQueue.Count > this._config.CacheQueueMax)
            {
                //等待最多100毫秒重新检查一次
                if (this._messageDequeueHandler.WaitOne(this._config.PushTimeoutMilliseconds))
                {
                    break;
                }
                else
                {
                    throw new TimeoutException("推送消息到缓存队列中超时,生产速度大于了消费速度");
                }
            }

            if (this._disposed)
            {
                throw new ObjectDisposedException(this.ToString());
            }

            this._messageQueue.Enqueue(message);
            this._messageEnqueueHandler.Set();
        }


        private void PushThreadMethod(ThreadPara threadPara)
        {
            try
            {
                const int interval = 1000;//最多等待1秒就强制执行一次
                IBMQDataMessage message;
                while (!threadPara.Token.IsCancellationRequested)
                {
                    try
                    {
                        try
                        {
                            this._messageEnqueueHandler.WaitOne(interval);
                        }
                        catch (ObjectDisposedException)
                        {
                            break;
                        }

                        while (this._messageQueue.TryDequeue(out message))
                        {
                            this.PushMessage(message);
                            this.PushCompleted?.Invoke(message);

                            if (!this._disposed)
                            {
                                this._messageDequeueHandler.Set();
                            }
                        }
                    }
                    catch (Exception exi)
                    {
                        ZLoger.Error(exi);
                    }
                }
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex);
            }
        }

        private readonly ArraySpan<byte> _pushMessageBuffer = new ArraySpan<byte>(new byte[8192]);
        private void PushMessage(IBMQDataMessage message)
        {
            switch (this._config.Protocol)
            {
                case NetProtocols.Tcp:
                    HashSet<IClient> clients;
                    if (this._tcpSubTopicClientDic.TryGetValue(message.Topic, out clients))
                    {
                        message.ToTLVBytes(this._pushMessageBuffer);
                        foreach (var client in clients)
                        {
                            ((NetTcpClient)client).Send(this._pushMessageBuffer.Array, this._pushMessageBuffer.Offset, this._pushMessageBuffer.Length);
                        }
                        message.Close();
                    }
                    break;
                case NetProtocols.Udp:
                    HashSet<EndPoint> endpoints;
                    if (this._udpSubTopicEndPointDic.TryGetValue(message.Topic, out endpoints))
                    {
                        message.ToBytes(this._pushMessageBuffer);
                        foreach (EndPoint endpoint in endpoints)
                        {
                            this._udpClient.SendTo(this._pushMessageBuffer.Array, this._pushMessageBuffer.Offset, this._pushMessageBuffer.Length, System.Net.Sockets.SocketFlags.None, endpoint);
                        }
                        message.Close();
                    }
                    break;
                default:
                    throw new NotSupportedException($"不支持协议{this._config.Protocol.ToString()}");
            }
        }

        /// <summary>
        /// 对象释放标识[true:已释放;fasle:未已释放]
        /// </summary>
        protected bool _disposed = false;
        /// <summary>
        /// Dispose
        /// </summary>
        public void Dispose()
        {
            if (this._disposed)
            {
                return;
            }
            this._disposed = true;

            try
            {
                this._unpacker.Dispose();

                this._heartThread.Stop();
                this._heartThread.Dispose();

                this._messageDequeueHandler.Set();
                ZLoger.Info($"BMQ推送线程停止为同步停止，超时时长为100秒");
                this._pushThread.SyncStop(100000);//同步取消，超时时长为100秒
                this._messageEnqueueHandler.Set();
                this._messageEnqueueHandler.Dispose();
                this._pushThread.Dispose();
                this._messageDequeueHandler.Dispose();

                lock (this._lock)
                {
                    switch (this._config.Protocol)
                    {
                        case NetProtocols.Tcp:
                            foreach (IClient client in _tcpClientSubTopicDic.Keys)
                            {
                                ((NetTcpClient)client).ConnectChanged -= TcpClient_ConnectChanged;
                                client.Dispose();
                            }
                            this._tcpSubTopicClientDic.Clear();
                            this._tcpClientSubTopicDic.Clear();
                            this._netTcpListener.Stop();
                            this._netTcpListener.Dispose();
                            break;
                        case NetProtocols.Udp:
                            this._udpSubTopicEndPointDic.Clear();
                            this._udpEndPointSubTopicDic.Clear();
                            this._udpClient.Dispose();
                            break;
                        default:
                            throw new NotSupportedException($"不支持协议{this._config.Protocol.ToString()}");
                    }
                }

                this._waitHandleManager.Dispose();
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex);
            }
        }

    }
}
